package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.model.common.constants.HotArticleConstants;
import com.heima.model.mess.app.ArticleVisitStreamMess;
import com.heima.model.mess.app.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * @Description:
 * @Version: V1.0
 */
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {

    // 接收消息 topic
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
    }

    // 流式处理成功后 发送消息 topic
    @Override
    public String sendTopic() {
        return HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
       return stream.flatMapValues(new ValueMapper<String, Iterable<?>>() {
            @Override
            public Iterable<?> apply(String value) {
                // value 发送消息UpdateArticleMess
                UpdateArticleMess articleMess = JSON.parseObject(value, UpdateArticleMess.class);
                log.info("receive message value:{}",value);
                // 按照文章的ID和文章行为类型计算
                return Arrays.asList(articleMess.getArticleId() + ":" + articleMess.getType());
            }
        })
        // 根据value进行分组
        .groupBy((key,value)->value)
        // 聚合计算时间间隔
        .windowedBy(TimeWindows.of(5000))
        // 聚合查询：求单词总个数
        .count()
        // 转成 KStream
        .toStream()
        // 处理后结果key和value转成string
        .map((key, value) -> {
            //key-->mess.getArticleId()+":"+mess.getType()
            //value  long类型  count之后的数值
            log.info("处理后结果key:{}-----value:{}",key.key().toString(),value.toString());
            return new KeyValue<>(key.key().toString(), formatObj(key.key().toString(), value.toString()));
        });
    }


    /**
     * 根据key的值来封装处理成功后的各个行为下的分值
     * @param key  mess.getArticleId()+":"+mess.getType()
     * @param value long类型  count之后的数值
     * @return
     */
    private String formatObj(String key, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        String[] split = key.split(":");
        mess.setArticleId(Long.valueOf(split[0]));

        if (UpdateArticleMess.UpdateArticleType.LIKES.name().equals(split[1])) {
            mess.setLike( Long.valueOf(value) );
        }
        if (UpdateArticleMess.UpdateArticleType.VIEWS.name().equals(split[1])) {
            mess.setView( Long.valueOf(value) );
        }
        if (UpdateArticleMess.UpdateArticleType.COMMENT.name().equals(split[1])) {
            mess.setComment( Long.valueOf(value) );
        }
        if (UpdateArticleMess.UpdateArticleType.COLLECTION.name().equals(split[1])) {
            mess.setCollect( Long.valueOf(value) );
        }

        log.info("formatObj aggs score ArticleVisitStreamMess:{}", mess);

        return JSON.toJSONString(mess);
    }
}